package com.zhouq.spark

import java.net.URL

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 求每门课程最受欢迎老师TopN  --2
  *
  */
object GroupFavTeacher2 {
  def main(args: Array[String]): Unit = {


    //前 N
    val topN = args(1).toInt

    //学科集合
    val subjects = Array("bigdata", "javaee", "php")

    val conf = new SparkConf().setAppName("FavTeacher").setMaster("local[4]")

    //创建spark 执行入口
    val sc = new SparkContext(conf)

    //指定读取数据
    val lines: RDD[String] = sc.textFile(args(0))

    val subjectTeacherAndOne: RDD[((String, String), Int)] = lines.map(line => {
      val index = line.lastIndexOf("/")
      var teacher = line.substring(index + 1)
      var httpHost = line.substring(0, index)
      var subject = new URL(httpHost).getHost.split("[.]")(0)
      ((subject, teacher), 1)
    })

    //将学科,老师联合当做key
    val reduced: RDD[((String, String), Int)] = subjectTeacherAndOne.reduceByKey(_ + _)



    //分组排序(按学科进行排序)
    //[学科,该学科对应的老师的数据]
    //    val grouped: RDD[(String, Iterable[((String, String), Int)])] = reduced.groupBy(_._1._1)
    //
    //    /**
    //      * 分组后,一个分区可能有多个学科的数据,一个学科就是一个迭代器
    //      * 将每一个分区的数据拿出来进行操作
    //      * 下面为什么可以使用scala 的sortBy 方法,是因为数据全部已经在一个机器上的scala 集合里面了.
    //      */
    //
    //    val sorted: RDD[(String, List[((String, String), Int)])] = grouped.mapValues(_.toList.sortBy(_._2).reverse.take(3))
    //
    //    //收集计算结果
    //    val result = sorted.collect()
    //
    //    println(result.toBuffer)

    /**
      * 先对学科进行过滤,然后再进行排序,调用RDD 的sortBy进行排序,避免scala 的排序当数据量大时,内存不足的情况.
      * take 是Action 操作,每次take 都会进行一次任务提交,具体查看日志打印情况
      */
    for (sub <- subjects) {
      //过滤出当前的学科
      val filtered: RDD[((String, String), Int)] = reduced.filter(_._1._1 == sub)

      //使用RDD 的 sortBy ,内存+磁盘排序,避免scala 中的排序因内存不足导致异常情况.
      //take 是Action 的,所以每次循环都会触发一次提交任务,祥见日志打印情况
      val favTeacher: Array[((String, String), Int)] = filtered.sortBy(_._2, false).take(topN)

      println(favTeacher.toBuffer)

    }

    sc.stop()
  }
}
